#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RocketMQ 有序消费者示例
演示如何使用有序消费模式
"""

import time
import json
import logging
from rocketmq.client import PushConsumer, ConsumeStatus, ReceivedMessage
from config import ROCKETMQ_CONFIG, MESSAGE_CONFIG, LOGGING_CONFIG

# 配置日志
logging.basicConfig(
    level=getattr(logging, LOGGING_CONFIG["level"]), 
    format=LOGGING_CONFIG["format"]
)
logger = logging.getLogger(__name__)


class OrderlyMessageHandler:
    """有序消息处理器"""
    
    def __init__(self):
        self.message_count = 0
        self.start_time = time.time()
        self.queue_message_count = {}  # 记录每个队列的消息数量
        self.last_message_time = {}    # 记录每个队列最后消息时间
    
    def __call__(self, msg: ReceivedMessage):
        """消息处理回调"""
        try:
            self.message_count += 1
            current_time = time.time()
            elapsed_time = current_time - self.start_time
            
            # 获取队列ID（兼容性处理）
            queue_id = getattr(msg, 'queue_id', 'unknown')
            
            # 更新队列统计
            if queue_id not in self.queue_message_count:
                self.queue_message_count[queue_id] = 0
            self.queue_message_count[queue_id] += 1
            
            # 计算队列内消息间隔
            queue_interval = 0
            if queue_id in self.last_message_time:
                queue_interval = current_time - self.last_message_time[queue_id]
            self.last_message_time[queue_id] = current_time
            
            logger.info(f"=== 有序消息 #{self.message_count} (队列: {queue_id}, 运行时间: {elapsed_time:.1f}s) ===")
            
            # 基本信息
            logger.info(f"  - Topic: {msg.topic}")
            logger.info(f"  - Tags: {msg.tags}")
            logger.info(f"  - Keys: {msg.keys}")
            logger.info(f"  - MsgId: {msg.id}")
            logger.info(f"  - QueueId: {queue_id}")
            logger.info(f"  - 队列内消息数: {self.queue_message_count[queue_id]}")
            logger.info(f"  - 队列内间隔: {queue_interval:.3f}s")
            
            # 兼容性处理：offset 可能不存在
            if hasattr(msg, 'offset'):
                logger.info(f"  - Offset: {msg.offset}")
            
            # 消息内容
            try:
                body = msg.body.decode(MESSAGE_CONFIG["encoding"])
                logger.info(f"  - 消息内容: {body}")
                
                # 尝试解析 JSON
                try:
                    data = json.loads(body)
                    if isinstance(data, dict):
                        logger.info(f"  - 消息ID: {data.get('id', 'N/A')}")
                        logger.info(f"  - 消息类型: {data.get('data', {}).get('type', 'N/A')}")
                        logger.info(f"  - 时间戳: {data.get('timestamp', 'N/A')}")
                        logger.info(f"  - 发送类型: {data.get('send_type', 'N/A')}")
                        
                        # 检查是否是有序消息
                        if data.get('send_type') == 'orderly':
                            logger.info(f"  - ✓ 这是有序消息")
                        else:
                            logger.info(f"  - ⚠ 这是普通消息")
                            
                except json.JSONDecodeError:
                    pass  # 不是 JSON 格式，忽略
                    
            except Exception as e:
                logger.warning(f"  - 消息内容解码失败: {e}")
                logger.info(f"  - 原始内容: {msg.body}")
            
            # 模拟有序处理时间
            process_time = 0.1  # 100ms
            time.sleep(process_time)
            
            logger.info(f"=== 有序消息处理完成 (耗时: {process_time:.3f}s) ===")
            
            return ConsumeStatus.CONSUME_SUCCESS
            
        except Exception as e:
            logger.error(f"有序消息处理失败: {e}")
            return ConsumeStatus.RECONSUME_LATER


class OrderlyRocketMQConsumer:
    """有序 RocketMQ 消费者类"""
    
    def __init__(self):
        self.consumer = None
        self.message_handler = OrderlyMessageHandler()
        self.namesrv_addr = ROCKETMQ_CONFIG["namesrv_addr"]
        self.access_key = ROCKETMQ_CONFIG["access_key"]
        self.secret_key = ROCKETMQ_CONFIG["secret_key"]
        self.group_name = ROCKETMQ_CONFIG["consumer_group"] + "_orderly"
        self.topic = ROCKETMQ_CONFIG["default_topic"]
        self.message_tags = ROCKETMQ_CONFIG["consumer_config"]["message_tags"]
    
    def start(self):
        """启动有序消费者"""
        try:
            logger.info("=== 启动有序 RocketMQ 消费者 ===")
            logger.info(f"NameServer: {self.namesrv_addr}")
            logger.info(f"消费者组: {self.group_name}")
            logger.info(f"主题: {self.topic}")
            logger.info(f"消息标签: {self.message_tags}")
            logger.info(f"有序消费: True")
            
            # 创建有序消费者
            self.consumer = PushConsumer(
                self.group_name,
                orderly=True  # 启用有序消费
            )
            self.consumer.set_name_server_address(self.namesrv_addr)
            self.consumer.set_session_credentials(self.access_key, self.secret_key, "ALIYUN")
            
            # 订阅主题
            self.consumer.subscribe(self.topic, self.message_handler, self.message_tags)
            
            # 启动消费者
            self.consumer.start()
            logger.info("✓ 有序消费者启动成功")
            logger.info("正在等待有序消息...")
            return True
            
        except Exception as e:
            logger.error(f"✗ 有序消费者启动失败: {e}")
            return False
    
    def run(self, timeout=60):
        """运行有序消费者"""
        try:
            start_time = time.time()
            
            while time.time() - start_time < timeout:
                time.sleep(1)
                
                # 每10秒打印一次状态
                elapsed = time.time() - start_time
                if int(elapsed) % 10 == 0 and elapsed > 0:
                    logger.info(f"有序消费者运行中... 已收到 {self.message_handler.message_count} 条消息")
                    
                    # 显示队列统计
                    if self.message_handler.queue_message_count:
                        logger.info("队列消息分布:")
                        for queue_id, count in self.message_handler.queue_message_count.items():
                            logger.info(f"  - 队列 {queue_id}: {count} 条消息")
            
            logger.info(f"有序消费者运行完成，共收到 {self.message_handler.message_count} 条消息")
            
        except KeyboardInterrupt:
            logger.info("用户中断有序消费者")
        except Exception as e:
            logger.error(f"有序消费者运行错误: {e}")
    
    def get_statistics(self):
        """获取统计信息"""
        return {
            "total_messages": self.message_handler.message_count,
            "queue_distribution": self.message_handler.queue_message_count,
            "run_time": time.time() - self.message_handler.start_time
        }
    
    def shutdown(self):
        """关闭有序消费者"""
        if self.consumer:
            self.consumer.shutdown()
            logger.info("有序消费者已关闭")


def main():
    """主函数"""
    logger.info("开始 RocketMQ 有序消费者演示...")
    logger.info("注意: 有序消费需要消息有相同的 MessageKey 才能保证顺序")
    
    # 创建有序消费者
    consumer = OrderlyRocketMQConsumer()
    
    try:
        # 启动有序消费者
        if not consumer.start():
            return
        
        # 运行有序消费者
        consumer.run(60)  # 运行60秒
        
        # 获取统计信息
        stats = consumer.get_statistics()
        
        # 总结
        logger.info("=== 有序消费演示总结 ===")
        logger.info(f"收到消息数量: {stats['total_messages']}")
        logger.info(f"运行时间: {stats['run_time']:.1f} 秒")
        
        if stats['queue_distribution']:
            logger.info("队列消息分布:")
            for queue_id, count in stats['queue_distribution'].items():
                logger.info(f"  - 队列 {queue_id}: {count} 条消息")
        
        if stats['total_messages'] > 0:
            logger.info("✓ 有序消费者演示成功完成")
            logger.info("提示: 有序消费确保同一队列内的消息按顺序处理")
        else:
            logger.warning("⚠ 未收到任何消息，请检查是否有消息发送到该主题")
            logger.info("提示: 可以运行 producer_demo.py 发送一些消息后再测试")
            
    except Exception as e:
        logger.error(f"演示过程中发生错误: {e}")
    finally:
        consumer.shutdown()


if __name__ == "__main__":
    main() 